feat: [EPIC-GW-05] 会话路由与流式透传中继 (Gateway 核心能力补全)#320
Conversation
为实现统一的流式会话路由奠定基础： 1. 身份注入：新增全局唯一的 ConnectionID 生成器，并注入至底层连接 Context 中，作为会话兜底标识。 2. 规则固化：在 JSON-RPC 解析层实现多级 Session 提取机制（显式字段优先 -> params 提取 -> 连接兜底 -> 缺失报错）。 3. 契约定义：新增 gateway.bindStream 请求体与 gateway.event 通知结构定义。
构建网关内部的核心消息分发枢纽： 1. 会话路由：维护 ConnectionID 到 Session/Run 的强类型多路复用映射表。 2. 自动续期：实现基于 TTL 的滑动窗口生命周期管理，并支持通过 gateway.ping 进行无参无感续期。 3. 容错隔离：为每个连接配置独立的有界发送队列 (Buffer)，并实装“慢连接自动剔除 (Slow Consumer Eviction)”机制，防止局部网络阻塞拖垮全局事件循环。
将所有物理入口接入单例中继器，实现外层协议抹平： 1. IPC 接入：实现命名管道/Unix Socket 的连接注册与解绑，复用串行写通道。 2. WS 接入：在 WebSocket 读写循环中接入中继队列，实现双向流式推送。 3. SSE 接入：严格遵循 Server-Sent Events 换行与协议规范，将 gateway.event 格式化输出。 4. 生命周期：在物理连接断开时，自动触发路由表的清理销毁，拒绝幽灵路由。
完成 EPIC-GW-05 的闭环组装： 1. 启动编排：在 Server 初始化阶段注入全局单例 StreamRelay。 2. 分发调度：拦截 gateway.bindStream 方法进行显式路由注册；拦截其余业务请求触发 TTL 自动续期。 3. 优雅关停：保障进程退出时先截断流式中继器，再关停底层网络，实现彻底的无泄漏退出。
作为流式中继功能的收尾，重构核心类型并补安全链路的边缘测试： 1. 类型与校验内聚：整理 types.go 与 validate.go，将流式路由特有的数据结构（如 StreamChannel, BindingHint）与参数校验规则集中管理，提升代码可读性。 2. 核心网络面断言：全面扩展 server_test.go, network_server_test.go 与 bootstrap_test.go，覆盖流式中继器在 IPC/WS/SSE 多端点下的并发注册、事件推送与优雅关停。 3. 协议与分发测试：增补 jsonrpc_test.go 与 rpc_dispatch_test.go，严格测试不同优先级下的 Session ID 提取规则及缺失时的标准 JSON-RPC 报错机制。 4. 边界漏洞封堵：通过 validate_additional_test.go 增加海量非标输入参数的防注入与越界测试。
|
/review |
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
![fennoai[bot]](https://avatars.githubusercontent.com/in/1924301?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
/simplify |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
ResultImplemented and pushed commit What I Changed
Comment Triage Outcome
Validation
Simplification PassRan a simplifier sub-agent review and applied safe simplification-compatible structure while preserving behavior in this patch scope. |
- start relay loops only after network listener is ready - stop relay during network server close for deterministic cleanup - reset relay start flags when loops exit and allow safe restart - cap per-connection stream bindings to prevent unbounded growth - add regression tests for lifecycle and limit paths Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: pionxe <148670367+pionxe@users.noreply.github.com>
|
@xgopilot 请你将覆盖率提升至百分之九十五以上 |
结果已补充并提交覆盖率测试，提交到当前 PR 分支： 已完成
覆盖率本地可复现口径：
未完成
|
Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: pionxe <148670367+pionxe@users.noreply.github.com>
|
/review |
| if s.relay == nil { | ||
| s.relay = NewStreamRelay(StreamRelayOptions{Logger: s.logger}) | ||
| } | ||
| s.relay.Start(ctx, runtimePort) |
There was a problem hiding this comment.
Serve starts relay background loops here, but Server.Close does not stop the relay. If callers use a long-lived context and invoke Close directly, relay cleanup/event goroutines can outlive the server and keep resources/subscriptions alive. Please add a symmetric relay shutdown in Close (or clearly document lifecycle ownership if intentional).
| continue | ||
| } | ||
| state.expireAt = now.Add(r.bindingTTL) | ||
| state.lastSeen = now |
There was a problem hiding this comment.
RefreshConnectionBindings assigns the same lastSeen timestamp to every binding on each ping. Combined with ResolveFallbackSessionID (which picks the max lastSeen), fallback session selection becomes non-deterministic when a connection has multiple active bindings, because map iteration order is random when timestamps tie. This can route subsequent session-less requests to different sessions across calls.
Consider preserving per-binding recency semantics (e.g., only refresh TTL on ping, not lastSeen), or introduce a deterministic tiebreaker / explicit "current fallback session" field.
| connectionBindingMap = make(map[bindingKey]*bindingState) | ||
| r.connectionBindings[normalizedConnectionID] = connectionBindingMap | ||
| } | ||
| if _, exists := connectionBindingMap[key]; !exists && len(connectionBindingMap) >= r.maxBindings { |
There was a problem hiding this comment.
maxBindings is checked using len(connectionBindingMap) before purging expired bindings. This can reject valid new binds (too many stream bindings) when the map is full of stale entries but cleanup tick hasn’t run yet. In practice this creates false negatives under bursty session churn.
Recommend pruning expired entries for this connection (or counting only active bindings) immediately before the limit check.
| if errors.Is(err, errFrameTooLarge) { | ||
| s.logger.Printf("decode frame failed: %v", err) | ||
| _ = s.writeRPCResponse(conn, encoder, protocol.NewJSONRPCErrorResponse( | ||
| _ = s.writeRPCResponseDirect(conn, encoder, protocol.NewJSONRPCErrorResponse( |
There was a problem hiding this comment.
这里在解析失败分支直接调用 writeRPCResponseDirect 写 socket，绕过了 relay 的单写协程；但同一连接上 relay writer 仍可能并发写（例如运行时事件/队列中已有响应），会破坏“单连接单写”约束，导致帧交错或 data race。建议将错误响应也统一入队，或在 direct write 与 writer 之间增加同一把写锁。
| s.relay = NewStreamRelay(StreamRelayOptions{Logger: s.logger}) | ||
| } | ||
| s.relay.Start(ctx, runtimePort) | ||
|
|
There was a problem hiding this comment.
Serve 在这里启动了 relay.Start(ctx, runtimePort)，但 Server.Close 没有对 relay 做对称 Stop。当调用方显式 Close 且 ctx 仍未取消时，cleanup/event loop 会继续存活，造成 goroutine 生命周期泄漏。建议在 Close 中补 relay.Stop()（与 NetworkServer.Close 一致）。
| if errors.Is(err, errFrameTooLarge) { | ||
| s.logger.Printf("decode frame failed: %v", err) | ||
| _ = s.writeRPCResponse(conn, encoder, protocol.NewJSONRPCErrorResponse( | ||
| _ = s.writeRPCResponseDirect(conn, encoder, protocol.NewJSONRPCErrorResponse( |
There was a problem hiding this comment.
这里直接写 conn（writeRPCResponseDirect）会绕过 relay 的单写协程。若同一连接上已有队列消息（例如 runtime 事件或其他异步响应）正在写出，会出现并发写同一 socket 的竞争，导致帧交错或写失败。建议统一走 relay 队列，或在此分支前确保 writer 已停并且不会再有并发发送。
| connectionBindingMap = make(map[bindingKey]*bindingState) | ||
| r.connectionBindings[normalizedConnectionID] = connectionBindingMap | ||
| } | ||
| if _, exists := connectionBindingMap[key]; !exists && len(connectionBindingMap) >= r.maxBindings { |
There was a problem hiding this comment.
maxBindings 检查发生在清理过期 binding 之前。若连接上存在已过期但尚未被清理的条目，这里会误判为超限并拒绝新绑定（too many stream bindings）。建议在容量判断前先剔除过期项，或容量统计仅计算未过期 binding。
| continue | ||
| } | ||
| state.expireAt = now.Add(r.bindingTTL) | ||
| state.lastSeen = now |
There was a problem hiding this comment.
RefreshConnectionBindings 给该连接下所有 binding 写入同一个 lastSeen 时间。之后 ResolveFallbackSessionID 以 lastSeen 选最近会话时，在多个 session 并存场景会退化为 map 迭代顺序（非确定）选择，导致无 session_id 请求的兜底会话不稳定。建议保留原始相对顺序或改为显式记录“当前活跃 session”。
3 participants
🎯 目标 (Motivation & Context)
本 PR 正式交付了网关层的终极核心基建：[EPIC-GW-05] 会话路由与流式中继。 在此之前，NeoCode 网关仅能处理“一问一答”的同步请求。本次重构彻底打通了从底层 Runtime Event 到前端连接（IPC/WS/SSE）的异步流式管道，使得大模型“打字机”实时输出、Agent 执行状态推送等复杂交互成为可能。
💡 核心架构决策：
id的 Notification (gateway.event) 形态。gateway.bindStream) + 隐式无感续期 (gateway.ping)”的智能生存策略，并结合 15 分钟 TTL 淘汰机制。Closes #318
✨ 主要变更 (Key Changes)
1. 核心中继器 (
stream_relay.go)2. 身份上下文提取 (
connection_context.go,protocol)ConnectionID体系，解决 HTTP 这种无状态请求的物理寻址问题。请求体显式声明 > URL params > 长连接兜底绑定 > 返回 missing_field。3. 多协议适配组装 (
server.go,network_server.go)event: \n data: \n\n的协议流格式，完美封装了 JSON-RPC notification。✅ 验收标准 (Acceptance Criteria)
gateway.bindStreamRPC 方法可成功建立会话映射。gateway.ping及后续带有 session 的请求均可正确触发该会话的 TTL 滑动续期。